package 离婚登记数和粗离婚率之比;

import org.apache.hadoop.mapreduce.Reducer;
import 全国粗离婚率.DivorceRateBean;
import 离婚登记数和粗离婚率之比.Bean.DivorceAnalysisBean;
import 离婚登记数和粗离婚率之比.Bean.DivorceAnalysisBean1;

import java.io.IOException;

public class DivorceAnalysisReducer extends Reducer<DivorceAnalysisBean, DivorceAnalysisBean1, DivorceAnalysisBean, DivorceAnalysisBean1> {
   DivorceAnalysisBean1 outValue = new DivorceAnalysisBean1();

    @Override
    protected void reduce(DivorceAnalysisBean key, Iterable<DivorceAnalysisBean1> values, Reducer<DivorceAnalysisBean, DivorceAnalysisBean1, DivorceAnalysisBean, DivorceAnalysisBean1>.Context context) throws IOException, InterruptedException {
        /**
         * 创建统计变量
         */
        float sumdivorceCount = 0;
        long sumpopulationCount = 0;
        double divorceRate = 0;

        // 遍历迭代器，累加
        for (DivorceAnalysisBean1 value : values) {
            sumdivorceCount += value.getDivorceCount();
            sumpopulationCount += value.getPopulationCount();
        }

        // 计算粗离婚率，需判断人口总数不为0，避免除0异常
        if (sumpopulationCount!= 0) {
            divorceRate = ((sumdivorceCount * 2) / sumpopulationCount) * 100;
        }else {
            divorceRate = 0 ;
        }

        // 使用String.format方法将离婚率结果保留两位小数
        String formatDivorceRate = String.format("%.2f", divorceRate);
        divorceRate = Double.parseDouble(formatDivorceRate);

        /**
         * 输出结果赋值，这里假设DivorceRateBean类有相应的方法来设置离婚率等属性，你可根据实际情况调整
         */
        outValue.set(sumdivorceCount,divorceRate);
        context.write(key, outValue); // 输出的是 K    V
    }
}
